package org.tao.lightningmq.client.event.spi.listener;

import com.alibaba.fastjson.JSON;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.tao.lightningmq.client.common.CommonCache;
import org.tao.lightningmq.client.event.model.TxMessageCallBackEvent;
import org.tao.lightningmq.common.dto.MessageDTO;
import org.tao.lightningmq.common.enums.BrokerEventCode;
import org.tao.lightningmq.common.enums.LocalTransactionState;
import org.tao.lightningmq.common.enums.TxMessageFlagEnum;
import org.tao.lightningmq.common.event.Listener;
import org.tao.lightningmq.common.tcp.TcpMsg;
import org.tao.lightningmq.common.transaction.TransactionListener;

import java.util.Objects;

public class TxMessageCallBackListener implements Listener<TxMessageCallBackEvent> {

    private static final Logger LOGGER = LoggerFactory.getLogger(TxMessageCallBackListener.class);

    @Override
    public void onReceive(TxMessageCallBackEvent event) throws Exception {
        MessageDTO messageDTO = event.getTxMessageCallbackReqDTO().getMessageDTO();
        Objects.requireNonNull(messageDTO.getProducerId(), "producerId is null");
        Objects.requireNonNull(messageDTO.getMsgId(), "msgId is null");
        TransactionListener transactionListener = CommonCache.getTransactionListenerMap().get(messageDTO.getProducerId());
        LocalTransactionState localTransactionState = transactionListener.callBackHandler(messageDTO);
        if(LocalTransactionState.COMMIT == localTransactionState) {
            messageDTO.setTxFlag(TxMessageFlagEnum.REMAIN_HALF_ACK.getCode());
            messageDTO.setLocalTxState(LocalTransactionState.COMMIT.getCode());
            TcpMsg remainHalfAckMsg = new TcpMsg(BrokerEventCode.PUSH_MSG.getCode(), JSON.toJSONBytes(messageDTO));
            event.getContext().writeAndFlush(remainHalfAckMsg);
            LOGGER.info("commit callback");
        } else if(LocalTransactionState.ROLLBACK == localTransactionState) {
            messageDTO.setTxFlag(TxMessageFlagEnum.REMAIN_HALF_ACK.getCode());
            messageDTO.setLocalTxState(LocalTransactionState.ROLLBACK.getCode());
            TcpMsg remainHalfAckMsg = new TcpMsg(BrokerEventCode.PUSH_MSG.getCode(), JSON.toJSONBytes(messageDTO));
            event.getContext().writeAndFlush(remainHalfAckMsg);
            LOGGER.info("rollback callback");
        }
    }
}
